package com.lefeng.cockpit.resource.storage.impl;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.lefeng.cockpit.common.enums.resource.ResourceType;
import com.lefeng.cockpit.resource.storage.StorageOperate;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.security.UserGroupInformation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.Closeable;
import java.io.IOException;
import java.io.InputStreamReader;
import java.nio.charset.StandardCharsets;
import java.security.PrivilegedExceptionAction;
import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;


/**
 * hadoop utils
 * single instance
 */
public class HadoopUtils implements Closeable, StorageOperate {

    public static final String FORMAT_S = "%s";
    public static final String FORMAT_S_S = "%s/%s";
    public static final String FOLDER_SEPARATOR = "/";
    public static final String RESOURCE_TYPE_FILE = "resources";


    private static final String RESOURCE_UPLOAD_PATH = "/cockpit";

    private static final Logger logger = LoggerFactory.getLogger(HadoopUtils.class);
    private String hdfsUser;
    private static final String HADOOP_UTILS_KEY = "HADOOP_UTILS_KEY";

    private static final LoadingCache<String, HadoopUtils> cache = CacheBuilder
            .newBuilder()
            .build(new CacheLoader<String, HadoopUtils>() {
                @Override
                public HadoopUtils load(String key) throws Exception {
                    return new HadoopUtils();
                }
            });


    private Configuration configuration;
    private FileSystem fs;

    private HadoopUtils() {
        hdfsUser = "root";
        init();
        initHdfsPath();
    }

    public static HadoopUtils getInstance() {
        return cache.getUnchecked(HADOOP_UTILS_KEY);
    }

    /**
     * init dolphinscheduler root path in hdfs
     */

    private void initHdfsPath() {
        Path path = new Path(RESOURCE_UPLOAD_PATH);
        try {
            if (!fs.exists(path)) {
                fs.mkdirs(path);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * init hadoop configuration
     */
    private void init() throws NullPointerException {
        try {
            configuration = new HdfsConfiguration();

            configuration.set("fs.defaultFS", "hdfs://172.16.11.70:8020");

            if (StringUtils.isNotEmpty(hdfsUser)) {
                UserGroupInformation ugi = UserGroupInformation.createRemoteUser(hdfsUser);
                ugi.doAs((PrivilegedExceptionAction<Boolean>) () -> {
                    fs = FileSystem.get(configuration);
                    return true;
                });
            } else {
                logger.warn("resource.hdfs.root.user is not set value!");
                fs = FileSystem.get(configuration);
            }

        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * @return Configuration
     */
    public Configuration getConfiguration() {
        return configuration;
    }


    @Override
    public boolean mkdir(String hdfsPath) throws IOException {
        return fs.mkdirs(new Path(hdfsPath));
    }

    @Override
    public String getFileName(ResourceType resourceType, String fileName) {
        return getHdfsFileName(resourceType, fileName);
    }

    @Override
    public boolean exists(String fileName) throws IOException {
        return fs.exists(new Path(fileName));
    }

    @Override
    public void createTenantDirIfNotExists() throws IOException {
        getInstance().mkdir(getHdfsResDir());
    }

    @Override
    public String getDir(ResourceType resourceType) {
        return getHdfsDir(resourceType);
    }

    @Override
    public boolean upload(String srcFile, String dstPath, boolean deleteSource, boolean overwrite) throws IOException {
        return copyLocalToHdfs(srcFile, dstPath, deleteSource, overwrite);
    }

    @Override
    public boolean delete(String hdfsFilePath, boolean recursive) throws IOException {
        return fs.delete(new Path(hdfsFilePath), recursive);
    }

    @Override
    public String getResDir() {
        return getHdfsResDir();
    }

    @Override
    public String getResourceFileName(String fullName) {
        return getHdfsResourceFileName(fullName);
    }

    @Override
    public List<String> vimFile(String filePath, int skipLineNums, int limit) throws IOException {
        return catFile(filePath, skipLineNums, limit);
    }


    /**
     * the src file is on the local disk.  Add it to FS at
     * the given dst name.
     *
     * @param srcFile      local file
     * @param dstHdfsPath  destination hdfs path
     * @param deleteSource whether to delete the src
     * @param overwrite    whether to overwrite an existing file
     * @return if success or not
     * @throws IOException errors
     */
    private boolean copyLocalToHdfs(String srcFile, String dstHdfsPath, boolean deleteSource,
                                    boolean overwrite) throws IOException {
        Path srcPath = new Path(srcFile);
        Path dstPath = new Path(dstHdfsPath);

        fs.copyFromLocalFile(deleteSource, overwrite, srcPath, dstPath);

        return true;
    }

    @Override
    public void close() throws IOException {
        if (fs != null) {
            try {
                fs.close();
            } catch (IOException e) {
                logger.error("Close HadoopUtils instance failed", e);
                throw new IOException("Close HadoopUtils instance failed", e);
            }
        }
    }


//     ========================================== private ===================================

    /**
     * get hdfs file name
     *
     * @param resourceType resource type
     * @param fileName     file name
     * @return hdfs file name
     */
    private static String getHdfsFileName(ResourceType resourceType, String fileName) {
        if (fileName.startsWith(FOLDER_SEPARATOR)) {
            fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
        }
        return String.format(FORMAT_S_S, getHdfsDir(resourceType), fileName);
    }

    /**
     * hdfs resource dir
     *
     * @param resourceType resource type
     * @return hdfs resource dir
     */
    private static String getHdfsDir(ResourceType resourceType) {
        String hdfsDir = "";
        if (resourceType.equals(ResourceType.FILE)) {
            hdfsDir = getHdfsResDir();
        }
        return hdfsDir;
    }

    /**
     * hdfs resource dir
     *
     * @return hdfs resource dir
     */
    private static String getHdfsResDir() {
        return String.format("%s/" + RESOURCE_TYPE_FILE, getHdfsTenantDir());
    }

    /**
     * @return file directory of tenants on hdfs
     */
    private static String getHdfsTenantDir() {
        return String.format(FORMAT_S, getHdfsDataBasePath());
    }

    /**
     * get data hdfs path
     *
     * @return data hdfs path
     */
    private static String getHdfsDataBasePath() {
        return RESOURCE_UPLOAD_PATH;
    }

    /**
     * cat file on hdfs
     *
     * @param hdfsFilePath hdfs file path
     * @param skipLineNums skip line numbers
     * @param limit        read how many lines
     * @return content of file
     * @throws IOException errors
     */
    private List<String> catFile(String hdfsFilePath, int skipLineNums, int limit) throws IOException {

        if (StringUtils.isBlank(hdfsFilePath)) {
            logger.error("hdfs file path:{} is blank", hdfsFilePath);
            return Collections.emptyList();
        }

        try (FSDataInputStream in = fs.open(new Path(hdfsFilePath))) {
            BufferedReader br = new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8));
            Stream<String> stream = br.lines().skip(skipLineNums).limit(limit);
            return stream.collect(Collectors.toList());
        }
    }

    /**
     * get absolute path and name for resource file on hdfs
     *
     * @param fileName file name
     * @return get absolute path and name for file on hdfs
     */
    private static String getHdfsResourceFileName(String fileName) {
        if (fileName.startsWith(FOLDER_SEPARATOR)) {
            fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
        }
        return String.format(FORMAT_S_S, getHdfsResDir(), fileName);
    }

}
